/*
 * (C) 2008 ASARI Takashi
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License (v2)
 * as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston,
 * MA 02111-1307 USA
 */

#include <limits.h>
#include <signal.h>

#include <event.h>
#include <evhttp.h>

#include <tcbdb.h>

#define DEBUG true

// Let's play in the global scope!
static TCBDB*           bdb          = NULL;
static struct evhttp*   httpd        = NULL;
static int              queue_size;

static struct evbuffer* EMPTY_BUFFER = NULL;

static void enqueue(struct evhttp_request* request, void* arg) {
  evhttp_add_header(request->output_headers, "Content-Type", "application/octet-stream");

  if (request->type != EVHTTP_REQ_POST) {
    evhttp_send_reply(request, 405, "Method not allowed (Only POST is available for enqueue.)", EMPTY_BUFFER);
    return;
  }

  // if length not provided -- return 411 Length required?
  // maybe handled by (modified; perhaps by asari in the future) libevent (http.c)

  const char* value  = EVBUFFER_DATA(request->input_buffer);
  const int   length = EVBUFFER_LENGTH(request->input_buffer);

  int ecode;

  if (tcbdbrnum(bdb) >= queue_size) {
    evhttp_send_reply(request, HTTP_SERVUNAVAIL, "Service Temporarily Unavailable (Queue full.)", EMPTY_BUFFER);
    return;
  }

  BDBCUR* cur = tcbdbcurnew(bdb);

  if (!tcbdbcurlast(cur)) {
    ecode = tcbdbecode(bdb);
    if (ecode == TCENOREC) {
      if (!tcbdbput(bdb, "", 0, value, length)) {
	evhttp_send_reply(request, 500, "Internal Server Error (Could not put the object to the queue.)", EMPTY_BUFFER);
	fprintf(stderr, "%s (line %d)\n", tcbdberrmsg(tcbdbecode(bdb)), __LINE__);
	tcbdbcurdel(cur);
	return;
      }
    } else {
      evhttp_send_reply(request, 500, "Internal Server Error (Could not move the cursor to the last of the queue.)", EMPTY_BUFFER);
      fprintf(stderr, "%s (line %d)\n", tcbdberrmsg(tcbdbecode(bdb)), __LINE__);
      tcbdbcurdel(cur); 
      return;
    }
  } else {
    if (!tcbdbcurput(cur, value, length, BDBCPAFTER)) {
      evhttp_send_reply(request, 500, "Internal Server Error (Could not append the object to the queue.)", EMPTY_BUFFER);
      fprintf(stderr, "%s (line %d)\n", tcbdberrmsg(tcbdbecode(bdb)), __LINE__);
      tcbdbcurdel(cur);
      return;
    }
  }

  tcbdbcurdel(cur);
  
  evhttp_send_reply(request, 204, "No Content (Enqueue succeeded.)", EMPTY_BUFFER);
}

static void dequeue(struct evhttp_request* request, void* arg) {  
  evhttp_add_header(request->output_headers, "Content-Type", "application/octet-stream");

  if (request->type != EVHTTP_REQ_POST) {
    evhttp_send_reply(request, 405, "Method not allowed (Only POST is available for dequeue.)", EMPTY_BUFFER);
    return;
  }

  if (tcbdbrnum(bdb) == 0) {
    evhttp_send_reply(request, HTTP_NOTFOUND, "Not Found (No object found in the queue.)", EMPTY_BUFFER);
    return;
  }

  BDBCUR* cur = tcbdbcurnew(bdb);

  if (!tcbdbcurfirst(cur)) {
    evhttp_send_reply(request, 500, "Internal Server Error (Could not move the cursor to the head of the queue.)", EMPTY_BUFFER);
    fprintf(stderr, "%s (line %d)\n", tcbdberrmsg(tcbdbecode(bdb)), __LINE__);
    tcbdbcurdel(cur);
    return;
  }

  int length;
  char* value = tcbdbcurval(cur, &length);

  if (value == NULL) {
    evhttp_send_reply(request, 500, "Internal Server Error (Could not get the object from the cursor.)", EMPTY_BUFFER);
    fprintf(stderr, "%s (line %d)\n", tcbdberrmsg(tcbdbecode(bdb)), __LINE__);
    tcbdbcurdel(cur);
    return;
  }

  if (!tcbdbcurout(cur)) {
    evhttp_send_reply(request, 500, "Internal Server Error (Could not delete the object from the queue.)", EMPTY_BUFFER);
    fprintf(stderr, "%s (line %d)\n", tcbdberrmsg(tcbdbecode(bdb)), __LINE__);
    tcbdbcurdel(cur);
    return;
  }    

  tcbdbcurdel(cur);

  // Tricky access
  struct evbuffer* buf = evbuffer_new();
  buf->orig_buffer = buf->buffer = value;
  buf->totallen = buf->off = length;

  evhttp_send_reply(request, 200, "OK (Dequeue succeeded.)", buf);

  // Don't forget that the return value from tcbdbcurval() has to free()'ed -- inside evbuffer_free() !
   evbuffer_free(buf);
}

/*
 * Print the status of the queue.
 * Most of the code were derived from tokyocabinet-1.3.1/tcbmgr.c
 * Copyright (C) 2006-2008 Mikio Hirabayashi / LGPL
 */
static void status(struct evhttp_request* request, void* arg) {

  evhttp_add_header(request->output_headers, "Content-Type", "text/plain");

  if (request->type != EVHTTP_REQ_GET) {
    evhttp_send_reply(request, 405, "Method not allowed (Only GET is available for status.)", EMPTY_BUFFER);
    return;
  }

  struct evbuffer* buf = evbuffer_new();

  const char *npath = tcbdbpath(bdb);
  if(!npath) npath = "(unknown)";
  evbuffer_add_printf(buf, "path: %s\n", npath);
  evbuffer_add_printf(buf, "database type: btree\n");
  uint8_t flags = tcbdbflags(bdb);
  evbuffer_add_printf(buf, "additional flags:");
  if(flags & BDBFOPEN) evbuffer_add_printf(buf, " open");
  if(flags & BDBFFATAL) evbuffer_add_printf(buf, " fatal");
  evbuffer_add_printf(buf, "\n");
  BDBCMP cmp = tcbdbcmpfunc(bdb);
  evbuffer_add_printf(buf, "comparison function: ");
  if (cmp == tcbdbcmplexical) {
    evbuffer_add_printf(buf, "lexical");
  } else if (cmp == tcbdbcmpdecimal) {
    evbuffer_add_printf(buf, "decimal");
  } else if (cmp == tcbdbcmpint32) {
    evbuffer_add_printf(buf, "int32");
  } else if (cmp == tcbdbcmpint64) {
    evbuffer_add_printf(buf, "int64");
  } else {
    evbuffer_add_printf(buf, "custom");
  }
  evbuffer_add_printf(buf, "\n");
  evbuffer_add_printf(buf, "max leaf member: %d\n", tcbdblmemb(bdb));
  evbuffer_add_printf(buf, "max node member: %d\n", tcbdbnmemb(bdb));
  evbuffer_add_printf(buf, "leaf number: %llu\n", (unsigned long long)tcbdblnum(bdb));
  evbuffer_add_printf(buf, "node number: %llu\n", (unsigned long long)tcbdbnnum(bdb));
  evbuffer_add_printf(buf, "bucket number: %llu\n", (unsigned long long)tcbdbbnum(bdb));
  if (bdb->hdb->cnt_writerec >= 0)
    evbuffer_add_printf(buf, "used bucket number: %lld\n", (long long)tcbdbbnumused(bdb));
  evbuffer_add_printf(buf, "alignment: %u\n", tcbdbalign(bdb));
  evbuffer_add_printf(buf, "free block pool: %u\n", tcbdbfbpmax(bdb));
  evbuffer_add_printf(buf, "inode number: %lld\n", (long long)tcbdbinode(bdb));
  char date[48];
  tcdatestrwww(tcbdbmtime(bdb), INT_MAX, date);
  evbuffer_add_printf(buf, "modified time: %s\n", date);
  uint8_t opts = tcbdbopts(bdb);
  evbuffer_add_printf(buf, "options:");
  if (opts & BDBTLARGE) evbuffer_add_printf(buf, " large");
  if (opts & BDBTDEFLATE) evbuffer_add_printf(buf, " deflate");
  if (opts & BDBTBZIP) evbuffer_add_printf(buf, " bzip");
  if (opts & BDBTTCBS) evbuffer_add_printf(buf, " tcbs");
  if (opts & BDBTEXCODEC) evbuffer_add_printf(buf, " excodec");
  evbuffer_add_printf(buf, "\n");
  evbuffer_add_printf(buf, "record number: %llu\n", (unsigned long long)tcbdbrnum(bdb));
  evbuffer_add_printf(buf, "file size: %llu\n", (unsigned long long)tcbdbfsiz(bdb));

  evhttp_send_reply(request, 200, "OK (Status succeeded.)", buf);

  evbuffer_free(buf);
};

static void bad_request_handler(struct evhttp_request *request, void* arg) {
  evhttp_send_reply(request, HTTP_BADREQUEST, "Bad Request (No such command)", EMPTY_BUFFER);
}

static void cleanup() {
  if (bdb != NULL) {
    char* path = tcstrdup(tcbdbpath(bdb));
    if (!tcbdbclose(bdb))
      fprintf(stderr, "Error on closing the queue (%s). Please try tcbdboptimize() on it.: %s\n", path, tcbdberrmsg(tcbdbecode(bdb)));
    else
      fprintf(stderr, "Queue (%s) successfully closed.\n", path);
    tcfree(path);
    tcbdbdel(bdb);
  }

  // event_loopbreak() breaks the event loop _after_ next event; when should I free following?
  // if (EMPTY_BUFFER != NULL) evbuffer_free(EMPTY_BUFFER);
  // if (httpd != NULL) evhttp_free(httpd);
}

static void signal_handler(int fd, short event, void *arg) {
  event_loopbreak();
  cleanup();
}

int main(int args, const char** argv) {
  // TODO: Validate args
  if (args != 5) {
    fprintf(stderr, "Usage: %s <queue-file> <size> <bind-address> <port>\n", argv[0]);
    return 1;
  }

  queue_size = atoi(argv[2]);

  const char* queue_name = argv[1];
  const char* bind_address = argv[3];
  int         bind_port = atoi(argv[4]);

  bdb = tcbdbnew();
  if (! tcbdbopen(bdb, queue_name, BDBOWRITER | BDBOCREAT | BDBOLCKNB)) {
    fprintf(stderr, "%s\n", tcbdberrmsg(tcbdbecode(bdb)));
    return 2;
  }
  
  event_init();

  static int signals[] = {
    SIGTERM, SIGINT, SIGHUP, SIGPIPE, SIGCHLD
  };
  static struct event signal_events[5];
  for (int i=0; i<5; i++) {
    event_set(&signal_events[i], signals[i], EV_SIGNAL, signal_handler, &signal_events[i]);
    event_add(&signal_events[i], NULL);
  }

  EMPTY_BUFFER = evbuffer_new();
  if (EMPTY_BUFFER == NULL) {
    fputs("failed to init EMPTY_REQUEST response buffer.\n", stderr);
    cleanup();
    return 4;
  }

  if (httpd = evhttp_start(bind_address, bind_port)) {
    evhttp_set_cb(httpd, "/enqueue", enqueue, NULL);
    evhttp_set_cb(httpd, "/dequeue", dequeue, NULL);
    evhttp_set_cb(httpd, "/status", status, NULL);
    evhttp_set_gencb(httpd, bad_request_handler, NULL);
    if (!event_dispatch()) return 5;
  } else {
    fputs("Could not start HTTP server.\n", stderr);
    return 6;
  }
}
